package service_management_logic

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"yunzhan/common/utils"
	agent "yunzhan/rpc-server/pb"

	"yunzhan/api-server/internal/svc"
	"yunzhan/api-server/internal/types"

	"github.com/zeromicro/go-zero/core/logx"
)

type ServiceOptionLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

// NewServiceOptionLogic 服务启停
func NewServiceOptionLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ServiceOptionLogic {
	return &ServiceOptionLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

// ServiceOption 服务启停操作
func (l *ServiceOptionLogic) ServiceOption(req *types.ServiceOperationRequest) (resp *types.DeployResponse, err error) {
	if req.Nodes == nil || len(req.Nodes) == 0 {
		return &types.DeployResponse{
			Code:    500,
			Message: "节点信息为空, 无法执行服务操作",
		}, errors.New("节点信息为空, 无法执行服务操作")
	}

	// 定义服务类型和服务文件名的映射关系
	serviceMap := map[string]string{
		"namenode":        "hadoop-namenode.service",
		"datanode":        "hadoop-datanode.service",
		"journalnode":     "hadoop-journalnode.service",
		"zkfailover":      "hadoop-zkfailovercontroller.service",
		"kafka":           "kafka.service",
		"zookeeper":       "zookeeper.service",
		"agent-server":    "agent-server.service",
		"redis":           "redis.service",
		"resourcemanager": "hadoop-resourcemanager.service",
		"nodemanager":     "hadoop-nodemanager.service",
		"grafana":         "grafana.service",
		"prometheus":      "prometheus.service",
		"alertmanager":    "alertmanager.service",
		"hmaster":         "hbase-master.service",
		"hregionserver":   "hbase-regionserver.service",
		"minio":           "minio.service",
		"hive-metastore":  "hive-metastore.service",
		"hive-server2":    "hive-server2.service",
	}

	// 将输入的服务类型转换为小写
	serviceType := strings.ToLower(req.ServiceType)

	// 获取服务名称
	serviceName, exists := serviceMap[serviceType]
	if !exists {
		return &types.DeployResponse{
			Code:    400,
			Message: fmt.Sprintf("无效的服务类型: %s", req.ServiceType),
		}, errors.New("无效的服务类型")
	}

	// 验证操作类型（启动或停止）
	action := strings.ToLower(req.OptionType)
	if action != "start" && action != "stop" {
		return &types.DeployResponse{
			Code:    400,
			Message: "无效的操作类型, 仅支持 'start' 或 'stop'",
		}, errors.New("无效的操作类型")
	}

	// 并发控制
	var wg sync.WaitGroup
	results := make(chan *NodeResult, len(req.Nodes))

	// 根据配置或节点数量动态设置并发度
	parallelism := l.svcCtx.Config.Parallelism
	if parallelism <= 0 || parallelism > len(req.Nodes) {
		parallelism = len(req.Nodes)
	}
	sem := make(chan struct{}, parallelism)

	// 遍历节点并执行操作
	for _, node := range req.Nodes {
		wg.Add(1)
		sem <- struct{}{}
		go func(n *agent.NodeInfo) {
			defer func() {
				<-sem
				wg.Done()
			}()

			// 每个节点的结果
			result := &NodeResult{
				Host:    n.Host,
				Success: true,
				Message: fmt.Sprintf("服务%s成功", action),
			}

			// 创建 SSH 连接池
			pool := utils.NewSSHConnectionPool(n.Host, n.Port, n.Username, n.Password, 5)
			client, err := pool.GetConnection()
			if err != nil {
				result.Success = false
				result.Message = fmt.Sprintf("获取 SSH 连接失败: %v", err)
				l.Logger.Errorf("节点: %s | 服务: %s | 操作: %s | 状态: 失败 | 描述: 获取 SSH 连接失败 | 错误详情: %v", n.Host, serviceName, action, err)
				results <- result
				return
			}
			defer pool.ReleaseConnection(client)

			// 创建 SSH 会话
			session, err := client.NewSession()
			if err != nil {
				result.Success = false
				result.Message = fmt.Sprintf("创建 SSH 会话失败: %v", err)
				l.Logger.Errorf("节点: %s | 服务: %s | 操作: %s | 状态: 失败 | 描述: 创建 SSH 会话失败 | 错误详情: %v", n.Host, serviceName, action, err)
				results <- result
				return
			}
			defer session.Close()

			// 拼接命令，防止命令注入
			command := fmt.Sprintf("/usr/bin/sudo systemctl %s %s", action, serviceName)

			// 执行命令
			output, err := session.CombinedOutput(command)
			if err != nil {
				result.Success = false
				result.Message = fmt.Sprintf("执行服务%s命令失败: %v, 输出: %s", action, err, string(output))
				l.Logger.Errorf("节点: %s | 服务: %s | 操作: %s | 状态: 失败 | 描述: 执行命令失败 | 错误详情: %v | 输出: %s", n.Host, serviceName, action, err, string(output))
				results <- result
				return
			}

			// 执行成功
			result.Message = fmt.Sprintf("服务%s成功", action)
			l.Logger.Infof("节点: %s | 服务: %s | 操作: %s | 状态: 成功 | 描述: 服务%s成功", n.Host, serviceName, action, action)
			results <- result
		}(node)
	}

	// 等待所有操作完成
	go func() {
		wg.Wait()
		close(results)
	}()

	// 收集操作结果
	var nodeResults []*NodeResult
	var successCount, failCount int
	var successMessages []string
	var failureMessages []string
	for result := range results {
		nodeResults = append(nodeResults, result)
		if result.Success {
			successCount++
			successMessages = append(successMessages, fmt.Sprintf("节点: %s | 状态: 成功 | 描述: %s", result.Host, result.Message))
		} else {
			failCount++
			failureMessages = append(failureMessages, fmt.Sprintf("节点: %s | 状态: 失败 | 描述: %s", result.Host, result.Message))
		}
	}

	// 构造返回信息
	var data []string
	if successCount > 0 {
		data = append(data, fmt.Sprintf("成功节点 (%d)", successCount))
		data = append(data, successMessages...)
	}
	if failCount > 0 {
		data = append(data, fmt.Sprintf("失败节点 (%d)", failCount))
		data = append(data, failureMessages...)
	}

	// 最终返回结果
	if failCount > 0 {
		return &types.DeployResponse{
			Code:    200,
			Message: fmt.Sprintf("服务操作完成: %d 个节点成功, %d 个节点失败。", successCount, failCount),
			Data:    data,
		}, nil
	}

	return &types.DeployResponse{
		Code:    200,
		Message: fmt.Sprintf("服务操作完成: 所有 %d 个节点操作成功！", len(req.Nodes)),
		Data:    data,
	}, nil
}

// NodeResult 用于保存单个节点的操作结果
type NodeResult struct {
	Host    string `json:"host"`
	Success bool   `json:"success"`
	Message string `json:"message"`
}
